package com.ylkj.bus.dispatcher.handler;

import com.hwl.datahub.DataHubWriteConfig;
import com.hwl.datahub.out.ProxyDatahubSinkFunctionStream;
import com.ylkj.bus.MessageContent;
import com.ylkj.bus.app.report.*;
import com.ylkj.bus.dispatcher.IStreamHandler;
import com.ylkj.dao.RdsConfig;
import com.ylkj.io.stream.datahub.SinkResolver;
import lombok.RequiredArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.util.OutputTag;

import java.util.List;

/**
 * @Author: wengzx
 * @Date: 2025/3/13
 * @Desc:
 */
@RequiredArgsConstructor
public class ReportAppStreamHandler implements IStreamHandler {
    private final RdsConfig rdsConfig;
    private final String rds1TableName;
    private final String rds2TableName;
    private final String rds4TableName;
    private final DataHubWriteConfig dataHubSinkConfig;
    @Override
    public void handle(DataStream<MessageContent> stream) {
        // 定义侧输出流标签
        OutputTag<List<IotmoAppDO>> outTagRds1 = new OutputTag<List<IotmoAppDO>>("outTagRds1") {};
        OutputTag<List<ContainerAppDO>> outTagRds2 = new OutputTag<List<ContainerAppDO>>("outTagRds2") {};

        // 主处理流程
        SingleOutputStreamOperator<MessageContent> sideoutProcess = stream
                .process(new ReportAppHandler(rdsConfig, outTagRds1, outTagRds2, rds4TableName))
                .name("report_app_sideout_process");

        // 获取侧输出流
        DataStream<List<IotmoAppDO>> sideOutput1 = sideoutProcess.getSideOutput(outTagRds1);
        DataStream<List<ContainerAppDO>> sideOutput2 = sideoutProcess.getSideOutput(outTagRds2);

        // 将侧输出流传递给处理类
        sideOutput1.process(new UnpackIotmoAppDeleteAndInsert(rds1TableName, rdsConfig)).name("unpack_iotmo_app_delete_and_insert");
        sideOutput2.process(new UnpackContainerAppDeleteAndInsert(rds2TableName, rdsConfig)).name("unpack_iotp_container_app_delete_and_insert");
        sideOutput1.addSink(new ProxyDatahubSinkFunctionStream<>(
                dataHubSinkConfig,
                new SinkResolver()
        )).name(dataHubSinkConfig.getTopicName() + "_datahub");
    }
}
